package teacher;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import pojo.Movie;
import system.MovieSystem;
import system.impl.MovieSystemImpl;
import tool.FileUtil;
import tool.LogCat;

import java.io.IOException;
import java.util.*;

// 2.根据评分数据得到每位用户的评分前三名（输出格式：用户编号,评分）
public class Score7{
    public static String InPath = "boy.txt";
    public static int keys = 1;

    public static class Score7Mapper extends Mapper<LongWritable, Text, IntWritable, Text> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //读取数据
            String line = value.toString();
            //进行处理
            //写入文件

            FileUtil.appendFile(InPath, line);


            context.write(new IntWritable(keys++), new Text(line));
        }
    }

    public static class Score7Reducer extends Reducer<IntWritable, Text, IntWritable, Text> {

        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            //尝试合成系统对象
            /*MovieSystem movieSystem = null;
            movieSystem = MovieSystemImpl.deserializeFromFile(InPath);
            if (movieSystem == null) {
                //进一步尝试
                //先保存文件内容
                String save = FileUtil.readFile(InPath);
                //写入换行
                FileUtil.writeFile(InPath, save+"\n");
                LogCat.line(13);
                LogCat.d3("写入->");
                LogCat.d3(save);
                //尝试合成
                movieSystem = MovieSystemImpl.deserializeFromFile(InPath);
                if( movieSystem == null) {
                    //还是空就没办法了
                    //写会save
                    FileUtil.writeFileCover(InPath, save);
                    LogCat.d3("覆盖写入");
                    return;
                }
            }

            LogCat.line(12);
            LogCat.line(12);
            LogCat.line(12);
            //如果成功
            //调用系统
            List<Movie> 男 = movieSystem.getTopMoviesBySex(3,"男");
            System.out.println(男);*/

            context.write(key, new Text(values.toString()));

        }
    }

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        //删除InPath
        FileUtil.writeFile(InPath,"");

        //MD,不演了
        LogCat.line();
        MovieSystem movieSystem = new MovieSystemImpl();
        if (movieSystem==null){
            LogCat.d3("系统初始化失败");
            return;
        }
        List<Movie> boy = movieSystem.getTopMoviesBySex(3, "男");
        LogCat.line();
        List<Movie> girl = movieSystem.getTopMoviesBySex(3, "女");

        LogCat.line();
        String boyStr = "";
        for (Movie movie : boy) {
            boyStr += "男";
            boyStr += ","+movie.getName();
            boyStr += ","+movie.getAverageScore();
            boyStr += "\n";
        }
        FileUtil.writeFile("boy.txt",boyStr);
        LogCat.line();

        String girlStr = "";
        for (Movie movie : girl) {
            girlStr += "女";
            girlStr += ","+movie.getName();
            girlStr += ","+movie.getAverageScore();
            girlStr += "\n";
        }
        FileUtil.writeFile("girl.txt",girlStr);

        LogCat.line();


        Configuration config = new Configuration();
        Job job = Job.getInstance(config);
        job.setJarByClass(Score7.class);
        job.setMapperClass(Score7Mapper.class);
        job.setReducerClass(Score7Reducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(Text.class);

        FileSystem fs = FileSystem.get(config);
        //读取系统路径
        Path inputPath = new Path(Config.SYSTEM_PATH);
        //输出路径
        Path outputPath = new Path(Config.OUT_PATH[7]);

        if (fs.exists(outputPath)) {
            fs.delete(outputPath, true);
        }

        FileInputFormat.setInputPaths(job, inputPath);
        FileOutputFormat.setOutputPath(job, outputPath);
        boolean isDone = job.waitForCompletion(true);
        System.out.println(isDone ? "执行成功" : "执行失败");
    }
}